package cn.doitedu.static_rule.entry;

import cn.doitedu.static_rule.beans.LogBean;
import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;

import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Properties;


/***
 * @author 涛哥
 * @nick_name "deep as the sea"
 * @contack qq:657270652 wx:doit_edu
 * @site www.51doit.cn
 * @date 2021/3/25
 * @desc 静态规则版运营系统
**/
public class Main {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "hdp01:9092,hdp02:9092,hdp03:9092");
        props.setProperty("auto.offset.reset", "latest");
        FlinkKafkaConsumer<String> eventSource = new FlinkKafkaConsumer<>("app_log", new SimpleStringSchema(), props);
        DataStreamSource<String> eventDataStream = env.addSource(eventSource);
        SingleOutputStreamOperator<LogBean> beanStream = eventDataStream.map(new MapFunction<String, LogBean>() {
            @Override
            public LogBean map(String value) throws Exception {
                return JSON.parseObject(value, LogBean.class);
            }
        });

        KeyedStream<LogBean, String> keyed = beanStream.keyBy(new KeySelector<LogBean, String>() {
            @Override
            public String getKey(LogBean value) throws Exception {
                return value.getDeviceId();
            }
        });

        SingleOutputStreamOperator<String> res = keyed.process(new KeyedProcessFunction<String, LogBean, String>() {
            ListState<LogBean> eventState;
            Connection hbaseConn;
            java.sql.Connection ckConn;
            Table profileTable;
            int cnt1;
            int cnt2;

            @Override
            public void open(Configuration parameters) throws Exception {
                ListStateDescriptor<LogBean> eventStateDesc = new ListStateDescriptor<>("events_state", LogBean.class);
                StateTtlConfig config = StateTtlConfig.newBuilder(Time.hours(2)).updateTtlOnReadAndWrite().build();
                eventStateDesc.enableTimeToLive(config);
                eventState = getRuntimeContext().getListState(eventStateDesc);

                org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();
                conf.set("hbase.zookeeper.quorum", "hdp01:2181,hdp02:2181,hdp03:2181");
                hbaseConn = ConnectionFactory.createConnection(conf);
                profileTable = hbaseConn.getTable(TableName.valueOf("user_profile"));

                final String ckDriver = "com.github.housepower.jdbc.ClickHouseDriver";
                final String ckUrl = "jdbc:clickhouse://192.168.77.63:9000/default";
                Class.forName(ckDriver);
                ckConn = DriverManager.getConnection(ckUrl);

            }

            @Override
            public void processElement(LogBean bean, Context ctx, Collector<String> out) throws Exception {
                eventState.add(bean);
                boolean flag = true;

                // 从state中计算事件次数
                Iterable<LogBean> logBeans = eventState.get();
                for (LogBean logBean : logBeans) {
                    if (logBean.getDeviceId().equals("e01") && logBean.getProperties().get("p1").equals("v1")) {
                        if (++cnt1 < 5) break;
                    }

                    if (logBean.getDeviceId().equals("e03") && logBean.getProperties().get("p2").equals("v2")) {
                        if (++cnt2 < 10) break;
                    }
                }

                // 从hbase查询画像数据
                Get get = new Get(bean.getDeviceId().getBytes());
                get.addColumn("f".getBytes(), "vip".getBytes());
                get.addColumn("f".getBytes(), "first_login".getBytes());
                Result result = profileTable.get(get);
                byte[] vipBytes = result.getValue("f".getBytes(), "vip".getBytes());
                byte[] firstLoginBytes = result.getValue("f".getBytes(), "first_login".getBytes());
                String vip = new String(vipBytes);
                String firstLogin = new String(firstLoginBytes);
                if (!vip.equals("vip2") || !firstLogin.equals("20")) {
                    flag = false;
                }

                // 从clickhouse查询次数
                if (flag) {
                    PreparedStatement pst = ckConn.prepareStatement("select count(1) from event_detail where deviceId = ? and eventId = ? and properties[?] = ?"
                            + " union all "
                            + "select count(1) from event_detail where deviceId = ? and eventId = ? and properties[?] = ?");

                    pst.setString(1, bean.getDeviceId());
                    pst.setString(2, "e01");
                    pst.setString(3, "p1");
                    pst.setString(4, "v1");

                    pst.setString(5, bean.getDeviceId());
                    pst.setString(6, "e03");
                    pst.setString(7, "p2");
                    pst.setString(8, "v2");

                    ResultSet resultSet = pst.executeQuery();
                    resultSet.next();
                    if (resultSet.getInt(1) < 5) flag = false;
                    resultSet.next();
                    if (resultSet.getInt(1) < 20) flag = false;

                }

                if (flag) {
                    out.collect("规则命中");
                }
            }
        });

        res.print();

        env.execute();
    }
}
